package com.taobao.ad.hadoop.main;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import com.taobao.ad.hadoop.common.JobNode;
 

public class HadoopTest1 extends JobNode {

	public static class Map extends MapReduceBase implements Mapper<Text, Text, Text, Text> {
		@Override
		public void map(Text key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
			output.collect(key, value);
		}

		@Override
		public void configure(JobConf job) {
			super.configure(job);
		}
	}

	public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> {
		@Override
		public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
			while (values.hasNext()) {
				output.collect(key, values.next());
			}
		}

		@Override
		public void configure(JobConf job) {
			super.configure(job);
		}
	}

	public static class Combine extends MapReduceBase implements Reducer<Text, Text, Text, Text> {
		@Override
		public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
		    while (values.hasNext()) {
				output.collect(key, values.next());
			}
		}

		@Override
		public void configure(JobConf job) {
			super.configure(job);
		}
	}

	public void ConfigJob() {
		super.ConfigJob(this.getClass(), Map.class, Reduce.class, Combine.class);
	}
}
